package sqlite_storage

import (
	"fmt"
	"go-iot-gateway/internal/apk/maths"
	"go-iot-gateway/internal/logger"
	"strings"
	"time"

	//"github.com/patrickmn/go-cache"
	storage_define "go-iot-gateway/internal/storage/define"
	"xorm.io/xorm"
)

type SqliteLatestStorage struct {
	running bool
	//通知队列大小
	//notifyValIds map[string]interface{}
	//数据库引擎
	dBEngine *xorm.Engine
	//写队列
	flushQueue chan storage_define.ValObject
	//刷盘配置
	flushConfig FlushConfig
	//写通知队列
	//notifyQueue chan storage_define.ValObject
	//订阅者
	//subscribeNotify map[string]<-chan storage_define.ValObject
}

// 初始化
// writeCacheSize 写缓存大小
func (storage *SqliteLatestStorage) init(flushConfig FlushConfig) {
	storage.running = true
	storage.flushConfig = flushConfig
	storage.flushConfig.FlushCacheSize = maths.MaxInt(360, flushConfig.FlushCacheSize)
	storage.flushQueue = make(chan storage_define.ValObject, storage.flushConfig.FlushCacheSize)
	//storage.notifyValIds = make(map[string]interface{}, storage.flushConfig.FlushCacheSize)
	//storage.notifyQueueSize = 500
	//storage.notifyQueue = make(chan storage_define.ValObject, storage.notifyQueueSize)
	//c := cache.New(5*time.Minute, 10*time.Minute)
	go func() {
		storage.syncSave()
	}()
}

func (storage *SqliteLatestStorage) SaveVal(val storage_define.ValObject) {
	if len(storage.flushQueue) < storage.flushConfig.FlushCacheSize {
		storage.flushQueue <- val
		return
	}
	logger.Storage().Warn("当前写队列已满 , 队列大小为 ", storage.flushConfig.FlushCacheSize)
}

func (storage *SqliteLatestStorage) NotifyVal(tagName string) {
	latestVal, err := storage.FindLatestVal(tagName)
	if err != nil {
		return
	}
	if latestVal == nil {
		return
	}
	storage.notify(latestVal)
}

func (storage *SqliteLatestStorage) notify(val storage_define.ValObject) {
	//todo 获取所有订阅者列表，并推送通知
	//if len(storage.notifyQueue) < storage.notifyQueueSize {
	//	storage.notifyQueue <- val
	//	return
	//}
	//logger.Storage().Warn("通知队列已满 , 队列大小为 ", storage.flushQueueSize)
}

func (storage *SqliteLatestStorage) SaveAndNotify(val storage_define.ValObject) {
	storage.SaveVal(val)
	//storage.notifyValIds[val.Id()] = nil
}

func (storage *SqliteLatestStorage) FixedVal(tagName string, val interface{}) {
	updateVal := new(LatestVal)
	updateVal.IsFix = true
	//todo 确定是否能这样更新
	_, err := storage.dBEngine.Where("tag_name=?", tagName).Update(updateVal)
	if err != nil {
		logger.Storage().Error("数据库更新失败 ", err)
	}
}

func (storage *SqliteLatestStorage) UnFixedVal(tagName string) {
	updateVal := new(LatestVal)
	updateVal.IsFix = false
	//todo 确定是否能这样更新
	_, err := storage.dBEngine.Where("tag_name=?", tagName).Update(updateVal)
	if err != nil {
		logger.Storage().Error("数据库更新失败 ", err)
	}
}

func (storage *SqliteLatestStorage) SubscribeNotify(subNo string) {
	//TODO implement me
	panic("implement me")
}

func (storage *SqliteLatestStorage) UnSubscribeNotify(subNo string) {
	//TODO implement me
	panic("implement me")
}

func (storage *SqliteLatestStorage) FindLatestVal(tagName string) (storage_define.ValObject, error) {
	latestVal := new(LatestVal)
	has, err := storage.dBEngine.Where("tag_name=?", strings.TrimSpace(tagName)).Get(latestVal)
	if err != nil {
		return nil, err
	}
	if !has {
		return nil, nil
	}
	return latestVal.ToValObject(), nil
}

func (storage *SqliteLatestStorage) FindLatestBySQL(query interface{}, args ...interface{}) []storage_define.ValObject {
	var rows []LatestVal
	err := storage.dBEngine.SQL(query, args...).Find(&rows)
	if err != nil {
		return make([]storage_define.ValObject, 0)
	}
	if rows == nil || len(rows) <= 0 {
		return make([]storage_define.ValObject, 0)
	}
	result := make([]storage_define.ValObject, 0, len(rows))
	for _, latestVal := range rows {
		result = append(result, latestVal.ToValObject())
	}
	return result
}

func (storage *SqliteLatestStorage) ExistsLatestVal(tagName string) bool {
	has, err := storage.dBEngine.SQL("select tag_name from "+storage_define.LATEST_TABLE_NAME+" where tag_name = ?",
		tagName).Exist()
	if err != nil {
		return false
	}
	return has
}

func (storage *SqliteLatestStorage) CloseStorage() {
	storage.running = false
	close(storage.flushQueue)
	//close(storage.notifyQueue)
	err := storage.dBEngine.Close()
	if err != nil {
		logger.Storage().Warn("配置库关闭失败")
	}
}

// 批量添加
func (storage *SqliteLatestStorage) syncSave() {
	flushSize := storage.flushConfig.FlushSize //批处理个数
	flushData := make([]storage_define.ValObject, 0, flushSize)
	flushInterval := time.Duration(storage.flushConfig.FlushIntervalMs)

	for storage.running {
		select {
		case <-time.After(time.Millisecond * flushInterval): //最小刷新间隔
			if len(flushData) > 0 {
				storage.batchUpdate(flushData)
				flushData = flushData[0:0] //清空
			}
		case val, ok := <-storage.flushQueue:
			if ok {
				flushData = append(flushData, val)
			}
			if len(flushData) >= flushSize {
				storage.batchUpdate(flushData)
				flushData = flushData[0:0] //清空
			}
		}

	}

}

func (storage *SqliteLatestStorage) batchUpdate(vals []storage_define.ValObject) {
	update_qos_sql := "update " + storage_define.LATEST_TABLE_NAME +
		" set qos = %d , updated_at = %d where tag_name='%s'; \n"
	update_num_val_sql := "update " + storage_define.LATEST_TABLE_NAME +
		" set qos=%d , updated_at = %d , num_val=%v , ts=%d , last_val = num_val  where tag_name='%s' and is_fix = 0; \n"
	update_str_val_sql := "update " + storage_define.LATEST_TABLE_NAME +
		" set qos=%d , updated_at = %d , str_val=%s , ts=%d , last_val = str_val where tag_name='%s' and is_fix = 0; \n"
	sql := ""

	updateSize := len(vals)
	logger.Storage().Info("批量更新 ", updateSize, " 条记录")
	for _, val := range vals {
		logger.Storage().Debug(val)
		if val.Val() == nil {
			sql += fmt.Sprintf(update_qos_sql, val.Qos(), time.Now().UnixMilli(), val.TagName())
		} else if _, ok := val.Val().(string); ok {
			sql += fmt.Sprintf(update_str_val_sql, val.Qos(), time.Now().UnixMilli(), val.Val(), val.Ts(), val.TagName())
		} else {
			sql += fmt.Sprintf(update_num_val_sql, val.Qos(), time.Now().UnixMilli(), val.Val(), val.Ts(), val.TagName())
		}
	}

	_, err := storage.dBEngine.Exec(sql)
	if err != nil {
		logger.Storage().Error("更新数据失败 ", err)
		//for _, val := range vals {
		//	delete(storage.notifyValIds, val.Id())
		//}
		return
	}

	//todo 如果有公式的tag , 要公式计算完后。再通知
	//for _, val := range vals {
	//	if _, ok := storage.notifyValIds[val.Id()]; ok {
	//		delete(storage.notifyValIds, val.Id())
	//		storage.notify(val)
	//	}
	//}
}

func (storage *SqliteLatestStorage) createVal(entity *LatestVal) (int64, error) {
	return storage.dBEngine.InsertOne(entity)
}

// 更新元数据
func (storage *SqliteLatestStorage) updateSchemaByTagName(entity *LatestVal) (int64, error) {

	sql := "update " + storage_define.LATEST_TABLE_NAME + " set data_type=? , channel_id=? , device_id=? where tag_name=?"

	r, err := storage.dBEngine.Exec(sql, entity.DataType, entity.ChannelId, entity.DeviceId, entity.TagName)

	if err != nil {
		return 0, err
	}

	return r.RowsAffected()
}

func (storage *SqliteLatestStorage) removeVal(tagName string) (int64, error) {
	sql := "delete from " + storage_define.LATEST_TABLE_NAME + " where tag_name=?"

	r, err := storage.dBEngine.Exec(sql, tagName)
	if err != nil {
		return 0, err
	}

	return r.RowsAffected()
}
